/*
 * core_mqtt.h
 *
 *  Created on: 2021年9月22日
 *      Author: LF
 */


#include <stdio.h>
#include <string.h>

#include "main.h"
#include "core_mqtt.h"




/*  mqtt 连接 broker  */
int mqtt_connect(char *host, int port, char *clientid, char *username, char *passwd)
{
	MQTTPacket_connectData		data = MQTTPacket_connectData_initializer;
	int							rv;
	unsigned char				buf[256];
	unsigned char				sessionPresent;
	unsigned char				connack_rc;

	if( !host || port<0 || !clientid)
	{
		printf("ERROR:INPUT FAILURE\r\n");
	}

	rv = transport_open(host, port);

	if( rv < 0 )
	{
		printf("SOCK CONNECT [%s:%d] failure, rv = %d\r\n", host, port, rv);
		return rv;
	}

	printf("SOCK CONNECT [%s:%d] ok\r\n", host, port);

	data.clientID.cstring = clientid;//客户端标识，用于区分每个客户端xxx为自定义，后面为固定格式
	data.keepAliveInterval = MQTT_KEEP_ALIVE_TIMEOUT_SECONDS;		//保活时间，单位为秒，也就是120秒内必须和服务器通讯一次，否则判定你下线，你就要重新连接
	data.cleansession = 1;//该标志置1服务器必须丢弃之前保持的客户端的信息，将该连接视为“不存在”

	if( username && passwd )
	{
		data.username.cstring = username;//用户名
		data.password.cstring = passwd;//密码

	}

	rv = MQTTSerialize_connect(buf, sizeof(buf), &data);/*1 构造连接报文*/
	if( rv < 0 )
	{
		printf("MQTTSerialize_connect failure, rv=%d\r\n", rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer mqtt_connect failure,rv=%d\r\n", rv);
		return -2;
	}

	HAL_Delay(800);

	memset(buf, 0, sizeof(buf));

	rv = MQTTPacket_read(buf, sizeof(buf), transport_getdata);

	if( CONNACK != rv )
	{
		printf("MQTTPacket_read for mqtt connack failure,rv=%d\n", rv);
		return -3;
	}

	if( (rv=MQTTDeserialize_connack(&sessionPresent, &connack_rc, buf, sizeof(buf))) != 1 || connack_rc != 0 )
	{
		printf("MQTTDeserialize_connack failure, rv=%d\n",rv);
		return -4;
	}

	printf("MQTT connected\r\n");
	return 0;
}

/*  断开连接  */
int mqtt_disconnect(void)
{
	int					rv;
	unsigned char		buf[256];

	rv = MQTTSerialize_disconnect(buf, sizeof(buf));
	if( rv < 0 )
	{
		printf("MQTTSerialize_disconnect failure,rv=%d\n", rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer mqtt_disconnect failure,rv=%d\r\n", rv);
		return -2;
	}


	return 0;
}

/*  订阅主题  */
int mqtt_subscribe_topic(char *topic, int qos, int msgid)
{
	MQTTString 				topicString = MQTTString_initializer;
	unsigned short 			submsgid;
	int 					subcount;
	int 					granted_qos;
	int						rv;
	unsigned char			buf[256];

	/* subscribe */
	topicString.cstring = topic;
	rv = MQTTSerialize_subscribe(buf, sizeof(buf), 0, msgid, 1, &topicString, &qos);
	if( rv < 0 )
	{
		printf("MQTTSerialize_subscribe failure,rv=%d\n",rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer for mqtt_subscribe_topic failure,rv=%d\n",rv);
		return -2;
	}

	HAL_Delay(800);

	memset(buf, 0, sizeof(buf));
	rv = MQTTPacket_read(buf, sizeof(buf), transport_getdata);

	if ( SUBACK != rv ) 	/* wait for suback */
	{

		printf("MQTTPacket_read for mqtt_suback failure,rv=%d\n",rv);
		return -3;

	}

	rv = MQTTDeserialize_suback(&submsgid, 1, &subcount, &granted_qos, buf, sizeof(buf));


	if (!rv || submsgid != msgid || granted_qos == 0x80)
	{
		printf("MQTTDeserialize_suback failure, rv=%d\n", rv);
		return -4;

	}



	return 0;


}

/*  取消订阅  */
int mqtt_unsubscribe_topic(char *topic, int msgid)
{
	MQTTString 				topicString = MQTTString_initializer;
	unsigned short 			submsgid;
	int						rv;
	unsigned char			buf[256];

	topicString.cstring = topic;
	rv = MQTTSerialize_unsubscribe(buf, sizeof(buf), 0, msgid, 1, &topicString);
	if( rv < 0 )
	{
		printf("MQTTSerialize_unsubscribe failure,rv=%d\n",rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer for mqtt_unsubscribe_topic failure,rv=%d\n",rv);
		return -2;
	}

	HAL_Delay(800);

	memset(buf, 0, sizeof(buf));
	rv = MQTTPacket_read(buf, sizeof(buf), transport_getdata);

	if ( UNSUBACK != rv ) 	/* wait for UNsuback */
	{

		printf("MQTTPacket_read for mqtt_unsuback failure,rv=%d\n",rv);
		return -3;

	}

	rv = MQTTDeserialize_unsuback(&submsgid, buf, sizeof(buf));


	if (submsgid != msgid || !rv )
	{
		printf("MQTTDeserialize_unsuback failure, rv=%d\n", rv);
		return -4;

	}


	return 0;
}

/*  发布消息  */
int mqtt_publish(char *topic, int qos,char *payload)
{
	MQTTString					topicString = MQTTString_initializer;
	int							rv;
	unsigned char				buf[256];
	unsigned char				dup = 0;
	unsigned char				retained = 0;
	unsigned short				packetid = 0;

	topicString.cstring = topic;

	rv = MQTTSerialize_publish(buf, sizeof(buf), dup, qos, retained, packetid, topicString, (unsigned char*)payload, strlen(payload));

	if( rv < 0 )
	{
		printf("MQTTSerialize_publish failure,rv=%d\r\n", rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer for mqtt_publish failure,rv=%d\n",rv);
		return -2;
	}

	HAL_Delay(800);

	memset(buf, 0, sizeof(buf));
	rv = MQTTPacket_read(buf, sizeof(buf), transport_getdata);

	if( PUBLISH != rv && -1 != rv )
	{
		printf("MQTTPacket_read for mqtt publish failure, rv=%d\r\n", rv);
		return -3;
	}


	return 0;
}

/*  保持心跳  */
int mqtt_pingreq(void)
{
	int				rv;
	unsigned char	buf[256];

	rv = MQTTSerialize_pingreq(buf, sizeof(buf));
	if( rv < 0 )
	{
		printf("MQTTSerialize_pingreq failure,rv=%d\r\n", rv);
		return -1;
	}

	if( rv != transport_sendPacketBuffer(buf, rv) )
	{
		printf("transport_sendPacketBuffer for mqtt_pingreq failure,rv=%d\n",rv);
		return -2;
	}

	HAL_Delay(800);

	memset(buf, 0, sizeof(buf));
	rv = MQTTPacket_read(buf, sizeof(buf), transport_getdata);

	if( PINGRESP != rv )
	{
		printf("MQTTPacket_read mqtt pingresp failure, rv=%d\r\n", rv);
		return -3;
	}


	return 0;
}





